package com.xinqing.bigdata.test.datastream

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

/**
  * @Author:CHQ
  * @Date:2020 /6/22 15:32
  * @Description
  */
class Myprocess() extends KeyedProcessFunction[String, SensorReading, String] {

  //获取上一次温度
  lazy val lastTemp: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))

  lazy val curTimerTs: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timerTs", classOf[Long]))

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {

    val preTemp = lastTemp.value()
    val curTemp = value.temperature
    val timeTs = curTimerTs.value()

    if (curTemp > preTemp && timeTs == 0.0) {
      val timeTs = ctx.timerService().currentProcessingTime() + 5000L
      //开启注册定时器
      ctx.timerService().registerProcessingTimeTimer(timeTs)
      lastTemp.update(curTemp)
      curTimerTs.update(timeTs)
    }
    if (curTemp < preTemp || preTemp == 0.0) {
      ctx.timerService().deleteProcessingTimeTimer(curTimerTs.value())
      curTimerTs.clear()
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    //发出告警
    out.collect("key：" + ctx.getCurrentKey + " 在5s内连续升温,请悉知")
    //以便从当前时刻开启下一次的告警
    curTimerTs.clear()
  }

}
